import lib.avro.Avro;

import java.io.File;
import java.io.InputStream;
import java.io.InputStreamReader;
// import java.io.ByteArrayInputStream;

import java.io.Writer;
import java.io.FileWriter;
import java.io.ByteArrayOutputStream;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileInputStream;

import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.HashMap;
import java.util.Base64;
import java.util.TimeZone;
import java.util.Date;
import java.util.Arrays;
import java.text.SimpleDateFormat;
import java.util.regex.Pattern;
import java.util.regex.Matcher;

import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;

import java.util.Collections;

import java.net.NetworkInterface;
import java.net.InetAddress;



import org.apache.commons.io.FileUtils;
import com.fasterxml.jackson.databind.ObjectMapper;

// import org.apache.commons.io.IOUtils;

import io.minio.MinioClient;

import io.minio.ListObjectsArgs;
import io.minio.PutObjectArgs;
import io.minio.GetObjectArgs;

import io.minio.Result;
import io.minio.messages.Item;

import org.apache.avro.generic.GenericContainer ;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.Encoder;

import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;

import org.apache.avro.io.ExtendedJsonDecoder ;
import com.github.luben.zstd.Zstd;

import io.github.cdimascio.dotenv.Dotenv;

import io.minio.ListenBucketNotificationArgs;
import io.minio.CloseableIterator;

import io.minio.messages.Event;
import io.minio.messages.NotificationRecords;

import org.apache.logging.log4j.Logger ;
import org.apache.logging.log4j.LogManager ;

public class BatchGate {
    private static final String MODULE = "push-gate-java:batch" ;
    private static final Logger LOGGER = LogManager.getLogger(LogManager.ROOT_LOGGER_NAME) ;
    private static void info(String message) {
        LOGGER.info("[component:{}] {}", MODULE, message) ;
    }
    private static void warn(String message) {
        LOGGER.warn("[component:{}] {}", MODULE, message) ;
    }
    private static void error(String message) {
        LOGGER.error("[component:{}] {}", MODULE, message) ;
    }


    private static String _resolveEnv(Dotenv dotenv, String name) {
        String v = dotenv.get(name);
        if (v == null)
            return null;
        Matcher p = Pattern.compile("\\$\\{([^}]*)\\}").matcher(v);

        String ret = v;
        while (p.find()) {
            String subVarValue = _resolveEnv(dotenv, p.group(1));
            if (subVarValue == null)
                return null;
            ret = v.replace(p.group(0), subVarValue);
        }
        return ret ;
    }

    public static String resolveEnv(Dotenv dotenv, String name, String defaultValue) {
        String resolveValue = _resolveEnv(dotenv, name);
        return resolveValue != null ? resolveValue : defaultValue;
    }
    public static String resolveMyIp(String nodes) throws Exception {
        for (NetworkInterface ni : Collections.list(NetworkInterface.getNetworkInterfaces())) {
            for (InetAddress addr : Collections.list(ni.getInetAddresses())) {
                String currentIp = addr.getHostAddress();
                if (nodes.contains(currentIp)) {
                    return currentIp;
                }
            }
        }
        return null;
    }

    static class Tuple2<A, B> {
        public A a;
        public B b;

        public Tuple2(A a, B b) {
            this.a = a;
            this.b = b;
        }
    }

    public static String parseChinaDt(String s) {
        long tsL = Long.parseLong(s);
        Date dt = new Date(tsL);
        SimpleDateFormat bjSdf = new SimpleDateFormat("yyyyMMdd");
        bjSdf.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
        return bjSdf.format(dt);
    }

    public static Map<String, String> restoreCategoryTsState(MinioClient client, String bucketName, String dir) throws Exception {
        Map<String, String> state = new HashMap<String, String>();
        ListObjectsArgs.Builder listArgsBuilder = ListObjectsArgs.builder().bucket(bucketName).recursive(false) ;
        if (dir != null) {
            listArgsBuilder.prefix(dir+"/");
        }
        for (Result<Item> categoryItemResults : client.listObjects(listArgsBuilder.build())) {
            String categoryPath = categoryItemResults.get().objectName();
            String categoryName = categoryPath.replaceAll("^.*/([^/]*)/", "$1") ;
            ListObjectsArgs categoryListArgs = ListObjectsArgs.builder().bucket(bucketName).prefix(categoryPath)
                    .recursive(true).build();
            for (Result<Item> fileItemResults : client.listObjects(categoryListArgs)) {
                String filePath = fileItemResults.get().objectName();
                filePath = URLDecoder.decode(filePath, StandardCharsets.UTF_8.toString());
                Matcher m = Pattern.compile("dt=\\d{8}/(\\d{8})_[^_]*_[^_]*_(\\d{4}).[^_]*_([^.]*)").matcher(filePath);
                if (!m.find()) {
                    continue;
                }
                String sourceCategoryName = categoryName + "_" + m.group(3);
                String currentTs = String.format("%s%s", m.group(1), m.group(2));
                String currentCategoryStateTs = state.get(sourceCategoryName);

                if (currentCategoryStateTs == null || currentTs.compareTo(currentCategoryStateTs) > 0) {
                    state.put(sourceCategoryName, currentTs);
                }
            }
        }
        return state;
    }
        
    public static List<Tuple2<String, String>> splitLine(GenericDatumReader<Object> reader, GenericDatumWriter<Object> writer, String line, boolean isJson, Schema schema)
            throws Exception {
        List<Tuple2<String, String>> collector = new ArrayList<Tuple2<String, String>>();

        Decoder decoder = null;

        GenericContainer avro = null ;
        if (isJson) {
            avro = Avro.fromJson(schema, line) ;
        } else {
            byte[] avro_zstd = Base64.getDecoder().decode(line.getBytes());
            avro = Avro.fromCompressedBin(schema, avro_zstd) ;
        }

        GenericData.Array<GenericRecord> sdkRecords = (GenericData.Array<GenericRecord>) avro ;
        for (GenericRecord sdkRecord : sdkRecords) {
            GenericRecord avroHeader = (GenericRecord) sdkRecord.get("common");

            GenericData.Array<GenericRecord> sdkEvents = (GenericData.Array<GenericRecord>) sdkRecord.get("events");
            for (GenericRecord sdkEvent : sdkEvents) {
                GenericRecord basic_info = (GenericRecord)sdkEvent.get("basic") ;
                String dt = parseChinaDt(basic_info.get("ts").toString());

                GenericData.Array<GenericRecord> sdkEventsOne = new GenericData.Array<GenericRecord>(
                        sdkEvents.getSchema(), Arrays.asList(sdkEvent));
                GenericRecord sdkRecordOne = new GenericData.Record(sdkRecord.getSchema());
                sdkRecordOne.put("events", sdkEventsOne);
                sdkRecordOne.put("common", avroHeader);

                if (sdkRecord.get("is_android") != null) {
                    sdkRecordOne.put("is_android", sdkRecord.get("is_android")) ;
                } else if (sdkRecord.get("is_ios") != null) {
                    sdkRecordOne.put("is_ios", sdkRecord.get("is_ios")) ;
                } else if (sdkRecord.get("is_web") != null) {
                    sdkRecordOne.put("is_web", sdkRecord.get("is_web")) ;
                } else if (sdkRecord.get("is_wechat") != null) {
                    sdkRecordOne.put("is_wechat", sdkRecord.get("is_wechat")) ;
                } else if (sdkRecord.get("is_h5") != null) {
                    sdkRecordOne.put("is_h5", sdkRecord.get("is_h5")) ;
                }

                GenericData.Array<GenericRecord> sdkRecordsOne = new GenericData.Array<GenericRecord>(schema,
                        Arrays.asList(sdkRecordOne));

                ByteArrayOutputStream os = new ByteArrayOutputStream();
                Encoder binEncoder = EncoderFactory.get().binaryEncoder(os, null);
                writer.write(sdkRecordsOne, binEncoder);
                binEncoder.flush();
                String newLine = Base64.getEncoder().encodeToString(Zstd.compress(os.toByteArray()));
                collector.add(new Tuple2<String, String>(dt, newLine));
            }
        }
        
        return collector;
    }

    public static void _listLocalFiles(File folder, List<String> collector) {
        for (File f : folder.listFiles()) {
            if (f.isDirectory()) {
                _listLocalFiles(f, collector);
            } else {
                collector.add(f.getAbsolutePath());
            }
        }
    }

    public static List<String> listLocalFiles(String filePath) {
        List<String> collector = new ArrayList<String>();
        _listLocalFiles(new File(filePath), collector);
        return collector;
    }

    public static List<String> listBufferFiles(MinioClient client, String bucketName, String dir, Map<String, String> state) throws Exception {
        List<String> collector = new ArrayList<String>();

        ListObjectsArgs.Builder listArgsBuilder = ListObjectsArgs.builder().bucket(bucketName).recursive(false) ;
        if (dir != null) {
            listArgsBuilder.prefix(dir+"/");
        }
        
        for (Result<Item> categoryItemResults : client.listObjects(listArgsBuilder.build())) {
            String categoryPath = categoryItemResults.get().objectName();
            String categoryName = categoryPath.replaceAll("(^.*/)?([^/]*)/", "$2") ;

            ListObjectsArgs categoryListArgs = ListObjectsArgs.builder().bucket(bucketName).prefix(categoryPath)
                    .recursive(true).build();

            String categoryStateTs = state.get(categoryName);
            for (Result<Item> fileItemResults : client.listObjects(categoryListArgs)) {
                String filePath = fileItemResults.get().objectName();
                filePath = URLDecoder.decode(filePath, StandardCharsets.UTF_8.toString());

                Matcher m = Pattern.compile("dt=(\\d{8})/[^_]*_[^_]*_(\\d{4})").matcher(filePath);
                if (!m.find()) {
                    continue;
                }

                String currentTs = String.format("%s%s", m.group(1), m.group(2));
                String currentCategoryStateTs = state.get(categoryName);

                if (currentCategoryStateTs == null || currentTs.compareTo(currentCategoryStateTs) > 0) {
                    state.put(categoryName, currentTs);
                }
                if (categoryStateTs == null || currentTs.compareTo(categoryStateTs) > 0) {
                    collector.add(filePath);
                }
            }
        }
        return collector;
    }

    public static void splitBufferFile(MinioClient inClient, String bucketName, String dir, String inPath, MinioClient outClient, String targetDb, String tmpDir)
            throws Exception {

        String partPath = inPath.replace(dir + "/", "");
        
        info("\ninPath:" + inPath);

        InputStream is = inClient.getObject(
            GetObjectArgs.builder().bucket(bucketName).object(inPath).build()
        );

        Schema schema = new Schema.Parser().parse(new File("meta-lib-static/avsc/dcSdkPlatformBulk.avsc"));
        GenericDatumReader<Object> reader = new GenericDatumReader<Object>(schema);
        GenericDatumWriter<Object> writer = new GenericDatumWriter<Object>(schema);


        Map<String, Writer> dtWriters = new HashMap<String, Writer>();

        
        boolean isJson = partPath.contains("json")? true : false;
        try (BufferedReader br = new BufferedReader(new InputStreamReader(is))) {
            String b64 = null;
            while ((b64 = br.readLine()) != null) {
                try {
                    for (Tuple2<String, String> tupleLine : splitLine(reader, writer, b64, isJson, schema)) {
                        String dt = tupleLine.a;
                        String b64New = tupleLine.b;
                        if (!dtWriters.containsKey(dt)) {
                            String filePath = (tmpDir + "/" + partPath).replaceAll(
                                    "/dt=(\\d{8})/",
                                    String.format("/dt=%s/$1_", dt));
                            File f = new File(filePath);
                            f.getParentFile().mkdirs();
                            Writer w = new BufferedWriter(new FileWriter(f)) ;
                            dtWriters.put(dt, w);
                        }
                        Writer w = dtWriters.get(dt);
                        w.write(b64New + "\n");
                    }
                } catch (Exception e) {
                    warn(String.format("splitBufferFile Exception [data: %s]", b64)) ;
                    e.printStackTrace();
                    continue;
                }
            }
            for (Writer wt : dtWriters.values()) {
                wt.close();
            }
            if (new File(tmpDir).exists()) {
                for (String filePath : listLocalFiles(tmpDir)) {
                    File inFile = new File(filePath) ;
                    InputStream in = new FileInputStream(inFile);
                    String outPath = filePath
                            .replace(tmpDir + "/",
                                    String.format("hive/warehouse/%s.db/", targetDb))
                            .replace("dc_sdk_push_avro", "dc_sdk_push").replace("dc_sdk_push_json", "dc_sdk_push");
                    long size = inFile.length() ;
                    info("outPath:" + outPath + "[size:" + size + "]");
                    outClient.putObject(PutObjectArgs.builder().bucket("user").object(outPath)
                            .stream(in, size, -1).build());
                    in.close();
                }
                FileUtils.deleteDirectory(new File(tmpDir));
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Dotenv dotenv = Dotenv.configure().directory("meta-lib-static/env/.env").load();

        String hosts= resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_HOSTS", "127.0.0.1");
        info("hosts:" + hosts);
        String myIp = resolveMyIp(hosts);
        if (myIp == null) {
            error("EXCHANGE_GATE_PUSH_BATCH_HOSTS not match! skip");
            System.exit(0);
        }
        info("myIp:" + myIp);

        String inMinioNodes = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_IN_MINIO_HOSTS", "127.0.0.1");
        String inMinioPort = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_IN_MINIO_PORT", "9000");
        String inMinioHost = String.format("http://%s:%s", inMinioNodes.split(",")[0], inMinioPort) ;
        info("in minio:" + inMinioHost);

        String inMinioUsername = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_IN_MINIO_USERNAME", "chinapex");
        String inMinioPassword = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_IN_MINIO_PASSWORD", "chinapex");

        String outMinioNodes = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_OUT_MINIO_HOSTS", "127.0.0.1");
        String outMinioPort = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_OUT_MINIO_PORT", "9000");
        String outMinioHost = String.format("http://%s:%s", outMinioNodes.split(",")[0], outMinioPort) ;
        info("out minio:" + outMinioHost);

        String outMinioUsername = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_OUT_MINIO_USERNAME", "chinapex");
        String outMinioPassword = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_OUT_MINIO_PASSWORD", "chinapex");
        String buffer_prefix = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_PREFIX", "data_buffer_dev");

        String tmp_dir = resolveEnv(dotenv, "EXCHANGE_GATE_PUSH_BATCH_TMP_DIR", "/tmp/exchange-buffer-minio-java") ;
        MinioClient inClient = MinioClient.builder().endpoint(inMinioHost)
                .credentials(inMinioUsername, inMinioPassword).build();

        MinioClient outClient = MinioClient.builder().endpoint(outMinioHost)
                .credentials(outMinioUsername, outMinioPassword)
                .build();

        String targetDb = buffer_prefix;
        String sourceBucketName = buffer_prefix.replace("_", "-");
        String sourceDir = null;

        Map<String, String> tsState = restoreCategoryTsState(outClient, "user",
                String.format("hive/warehouse/%s.db", targetDb));
        info("init tsState:" + tsState);

        info("refresh state =>");
        while (true) {
            try {
                for (String bufferPath : listBufferFiles(inClient, sourceBucketName, sourceDir, tsState)) {
                    splitBufferFile(inClient, sourceBucketName, sourceDir, bufferPath, outClient, targetDb, tmp_dir);
                }
            } catch (Exception e) {
                warn("MINIO process failed, retry...") ;
                e.printStackTrace();
                FileUtils.deleteDirectory(new File(tmp_dir));
                continue ;
            }
            info("tsState:" + tsState);
            info("sleep for 5 seconds =>");
            Thread.sleep(1000 * 5);
        }

    }

/*
    public static void notifyMain(String[] args) throws Exception {
        MinioClient minioClient =
            MinioClient.builder()
            .endpoint("http://10.25.21.40:9000")
            .credentials("chinapex", "chinapex")
            .build();

      String[] events = { "s3:ObjectCreated:*" };
      try (CloseableIterator<Result<NotificationRecords>> ci =
          minioClient.listenBucketNotification(
              ListenBucketNotificationArgs.builder()
                  .bucket("data-buffer-uat")
                  .prefix("")
                  .suffix("")
                  .events(events)
                  .build())) {
        while (ci.hasNext()) {
          NotificationRecords records = ci.next().get();
          Event event = records.events().get(0);
          System.out.println(event.bucketName() + "/" + event.objectName() + " has been created");
        }
      } catch (Exception e) {
        System.out.println("Error occurred: " + e);
      }
    }
*/
}
